package com.atguigu.day02;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

// FLATMAP：针对列表或者流中的每一个元素输出0个、1个或者多个元素
// 将Mary的数据复制输出
// 将Bob的数据直接输出
// 其他用户的数据过滤掉
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .flatMap(new MyFlatMapFunction())
                .print();

        env.execute();
    }

    public static class MyFlatMapFunction implements FlatMapFunction<ClickEvent, ClickEvent> {
        @Override
        public void flatMap(ClickEvent value, Collector<ClickEvent> out) throws Exception {
            if (value.username.equals("Mary")) {
                out.collect(value);
                out.collect(value);
            } else if (value.username.equals("Bob")) {
                out.collect(value);
            }
        }
    }
}
